package com.tc.vms.kafka;

import com.tc.vms.utils.VmsThreadFactory;
import kafka.common.TopicAndPartition;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerFetcherManager;
import kafka.consumer.PartitionTopicInfo;
import org.I0Itec.zkclient.ZkClient;

import java.util.HashMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by yonghua.zhang on 2016/5/16.
 */
public class KafkaZkConsumerConnector {
    private final ConsumerConfig config;
    private final AtomicBoolean isShuttingDown = new AtomicBoolean(false);
    private final Object rebalanceLock = new Object();
    private volatile ConsumerFetcherManager consumerFetcherManager;

    private volatile ZkClient zkClient = null;
    private final ConcurrentHashMap<String, HashMap<Integer, PartitionTopicInfo>> topicRegistry = new ConcurrentHashMap<String, HashMap<Integer, PartitionTopicInfo>>();
    private final HashMap<TopicAndPartition, Long> checkpointedZkOffsets = new HashMap<TopicAndPartition, Long>();
    //private val topicThreadIdAndQueues = new Pool[(String, ConsumerThreadId), BlockingQueue[FetchedDataChunk]]
    private final ExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new VmsThreadFactory("kafka-consumer-scheduler-"));
/*

	private volatile ZookeeperConsumerConnector.ZKSessionExpireListener sessionExpirationListener = null;
	private volatile ZKTopicPartitionChangeListener topicPartitionChangeListener = null;
	private volatile ZKRebalancerListener loadBalancerListener = null;

	private var offsetsChannel: BlockingChannel = null
	private final Object offsetsChannelLock = new Object();

	private var wildcardTopicWatcher: ZookeeperTopicEventWatcher = null

	// useful for tracking migration of consumers to store offsets in kafka
	private val kafkaCommitMeter = newMeter("KafkaCommitsPerSec", "commits", TimeUnit.SECONDS, Map("clientId" -> config.clientId))
	private val zkCommitMeter = newMeter("ZooKeeperCommitsPerSec", "commits", TimeUnit.SECONDS, Map("clientId" -> config.clientId))
	private val rebalanceTimer = new KafkaTimer(newTimer("RebalanceRateAndTime", TimeUnit.MILLISECONDS, TimeUnit.SECONDS, Map("clientId" -> config.clientId)))

*/


    public KafkaZkConsumerConnector(ConsumerConfig config) {
        this.config = config;
    }
}
